#include "sys.h"
#include "log.h"
#include "thread_pool.h"

#define FEATURE_HAS(tp, x) (tp->feature & x)

#define LIMIT_NUMBER 1000
#define MAX_TASK_NUMBER 1000
#define NORMAL_NUMBER 10

#define TASK_RUN(task) task->fun(task->arg)

static void _start_threads(thread_pool_t* tp);

int thread_pool_set_auto_extern(thread_pool_t* tp)
{
    tp->feature |= THREAD_POOL_AUTO_EXTERN;
    return tp->feature;
}
int thread_pool_set_limit_number(thread_pool_t* tp, int limit_number)
{
    tp->limit_number = limit_number;
    return tp->limit_number;
}
int thread_pool_set_max_task_number(thread_pool_t* tp, int max_task_number)
{
    tp->max_task_number = max_task_number;
    return tp->max_task_number;
}
int thread_pool_get_running_number(thread_pool_t* tp)
{
    return tp->running_number;
}
int thread_pool_get_task_number(thread_pool_t* tp)
{
    return tp->cur_task_number;
}
void thread_pool_shutdown(thread_pool_t* tp)
{
    pthread_mutex_lock(&tp->mutex);
    tp->shutdown = 1;
    pthread_mutex_unlock(&tp->mutex);
}
static void _thread_pool_extern_pool(thread_pool_t* tp, int extern_number)
{
    tp->extern_number = extern_number;
    tp->normal_number += extern_number;
    _start_threads(tp);
}
static void* _task_routine(void* arg)
{
    thread_pool_t* tp = (thread_pool_t*)arg;

    log_trace("pthread id(%u) start...", pthread_self());

    while (1) {
        pthread_mutex_lock(&tp->mutex);
        if (tp->running_number)
            tp->running_number--;

        if (tp->shutdown) {
            pthread_mutex_unlock(&tp->mutex);
            pthread_exit(0);
        }

        if (tp->task_list == NULL) {
            pthread_cond_wait(&tp->cond, &tp->mutex);
        }

        if(tp->task_list == NULL){
            pthread_mutex_unlock(&tp->mutex);
            continue;
        }

        thread_task_t* task = tp->task_list;
        tp->task_list = task->next;
        tp->cur_task_number--;
        tp->running_number++;
        pthread_mutex_unlock(&tp->mutex);

        TASK_RUN(task);
        free(task);
    }

    return NULL;
}
static void _start_threads(thread_pool_t* tp)
{
    int i = 0;
    pthread_t th_id;

    for (i = 0; i < tp->extern_number; i++) {
        pthread_create(&th_id, NULL, _task_routine, (void*)tp);
        pthread_detach(th_id);
    }
}
thread_pool_t* thread_pool_new(int normal_number)
{
    assert(normal_number >= 0);

    thread_pool_t* tp = malloc(sizeof(thread_pool_t));
    if (!tp) {
        log_syserr("new thread pool error.");
        return NULL;
    }
    memset(tp, 0, sizeof(thread_pool_t));
    tp->limit_number = LIMIT_NUMBER;
    tp->normal_number = normal_number ? normal_number : NORMAL_NUMBER;
    tp->max_task_number = MAX_TASK_NUMBER;
    tp->extern_number = tp->normal_number;
    pthread_mutex_init(&tp->mutex, NULL);
    pthread_cond_init(&tp->cond, NULL);
    _start_threads(tp);
    return tp;
}
static thread_task_t* _thread_task_new(thread_task_fun fun, void* arg)
{
    thread_task_t* task = malloc(sizeof(thread_task_t));
    if (!task) {
        log_syserr("new thread task error.");
        return NULL;
    }
    memset(task, 0, sizeof(thread_task_t));
    task->fun = fun;
    task->arg = arg;
    return task;
}
int thread_pool_add_task(thread_pool_t* tp, thread_task_fun fun, void* arg)
{
    thread_task_t* task = _thread_task_new(fun, arg);
    if (!task) {
        return 0;
    }
    pthread_mutex_lock(&tp->mutex);
    if(tp->cur_task_number > tp->max_task_number){
        log_error("Add task failed, too many tasks waiting; max task number is %d", tp->max_task_number);
        pthread_mutex_unlock(&tp->mutex);
        return 0;
    }
    if (tp->task_list == NULL) {
        tp->task_list = task;
    } else {
        thread_task_t* ptr = tp->task_list;
        while (ptr->next)
            ptr = ptr->next;

        ptr->next = task;
    }
    tp->cur_task_number++;
    if (FEATURE_HAS(tp, THREAD_POOL_AUTO_EXTERN)) {
        int extern_number = tp->cur_task_number + tp->running_number - tp->normal_number;
        if(extern_number > 0){
            extern_number = (tp->normal_number + extern_number > tp->limit_number) \
                                ? (tp->limit_number - tp->normal_number) : extern_number;
            _thread_pool_extern_pool(tp, extern_number);
        }
    }
    pthread_cond_signal(&tp->cond);
    pthread_mutex_unlock(&tp->mutex);

    return 1;
}

void thread_pool_destroy(thread_pool_t* tp)
{
    thread_task_t *ptr, *next;

    thread_pool_shutdown(tp);
    ptr = tp->task_list;
    while (ptr) {
        next = ptr->next;
        free(ptr);
        ptr = next;
    }
    memset(tp, 0, sizeof(thread_pool_t));
    free(tp);
    log_trace("thread pool destroied.");
}
